package com.ruoyi.project.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
@Slf4j
public class KafkaConsumer {

    //不指定group，默认取yml里配置的
//    @KafkaListener(topics = {"devices_cb6d8e7a1f_events"})
//    @KafkaListener(topicPattern = "devices_.*_events")
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord) {
        log.info("topic:{}", consumerRecord.topic());
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            String msg = optional.get().toString();
            log.info("message:{}", msg);

            try{

            }catch (Exception e){
                log.error("KafkaConsumer异常", e);
            }

        }
    }
}

